package com.example.springbootkafka.config;

import com.example.springbootkafka.kafka.OldDelayProcessorSupplier;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.processor.WallclockTimestampExtractor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
import org.springframework.kafka.config.TopicBuilder;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * kafka配置
 *
 * @author huangchangjin
 * @date 2025/1/8 9:55
 */
@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaConfig {
    public static final String STREAM_IN_TOPIC = "stream-in-topic";
    public static final String STREAM_OUT_TOPIC = "stream-out-topic";

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Bean("in_topic")
    public NewTopic streamingTopic1() {
        return genTopic(STREAM_IN_TOPIC);
    }

    @Bean("out_topic")
    public NewTopic streamingTopic2() {
        return genTopic(STREAM_OUT_TOPIC);
    }

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    KafkaStreamsConfiguration kStreamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        //props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());

        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public StreamsBuilderFactoryBeanConfigurer configurer() {
        return fb -> fb.setStateListener((newState, oldState) -> {
            System.out.println("State transition from " + oldState + " to " + newState);
        });
    }

    @Bean
    public KStream<String, String> kStream(StreamsBuilder kStreamBuilder) {
        KStream<String, String> stream = demo3(kStreamBuilder);

        return stream;
    }

    private KStream<String, String> demo1(StreamsBuilder kStreamBuilder) {
        KStream<String, String> stream = kStreamBuilder.stream(STREAM_IN_TOPIC);
        stream
                .mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(1_000)))
                .reduce((String value1, String value2) -> value1 + value2,
                        Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to(STREAM_OUT_TOPIC);

        stream.print(Printed.toSysOut());
        return stream;

    }

    private KStream<String, String> demo2(StreamsBuilder kStreamBuilder) {
        KStream<String, String> stream = kStreamBuilder.stream(STREAM_IN_TOPIC);
        stream
                //.mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(10000)))
                .reduce((String value1, String value2) -> value1 + value2,
                        Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to(STREAM_OUT_TOPIC);

        stream.print(Printed.toSysOut());
        return stream;

    }

    private KStream<String, String> demo3(StreamsBuilder kStreamBuilder) {
        KStream<String, String> stream = kStreamBuilder.stream(STREAM_IN_TOPIC);


        OldDelayProcessorSupplier delayProcessorSupplier = new OldDelayProcessorSupplier();
        stream.process(delayProcessorSupplier, Named.as("windowStore"));


        stream.to(STREAM_OUT_TOPIC);


        return stream;

    }

    private NewTopic genTopic(String topicName) {
        return TopicBuilder.name(topicName)
                .partitions(1)
                .replicas(1)
                .build();
    }


}
